package fox.framework.rpc.endpoint.channel.netty;

import fox.framework.rpc.client.message.request.RpcClientRequest;
import fox.framework.rpc.client.message.response.RpcClientResponse;
import fox.framework.rpc.client.node.RpcNode;
import fox.framework.rpc.common.ExceptionUtils;
import fox.framework.rpc.endpoint.channel.RpcChannel;
import fox.framework.rpc.endpoint.channel.RpcStatusChannel;
import fox.framework.rpc.endpoint.exchange.Future;
import fox.framework.rpc.endpoint.exchange.RpcExchange;
import fox.framework.rpc.exception.RpcRequestUndoException;
import fox.framework.rpc.exception.RpcRuntimeException;
import io.netty.channel.Channel;
import lombok.Setter;

import java.util.concurrent.TimeUnit;

/**
 * @author cuichao
 * @Description: RpcNettyChannel
 * @Date: create in 2020/12/29 22:25
 */
public class RpcNettyChannel implements RpcStatusChannel, RpcChannel.Sync {

    private RpcNode node;
    /**
     * netty channel channel.writeAndFlush(request).sync 只是阻塞到发送成功而不是响应返回
     */
    private RpcExchange sync;

    private Channel channel;

    public RpcNettyChannel(RpcNode node, Channel channel) {
        this.node = node;
        this.channel = channel;
    }

    @Override
    public RpcClientResponse request(RpcClientRequest request) throws RpcRuntimeException {
        //register sync future
        Future future = sync.register(request);
        //async send
        channel.writeAndFlush(request).addListener(f -> {
            if (!f.isSuccess()) {
                //cancel rpc request
                future.undo(f.cause());
            }
        });
        RpcClientResponse response = null;
        try {
            //blocking
            response = future.get(30,TimeUnit.SECONDS);
        } catch (RpcRequestUndoException e) {
            //取消操作
            throw ExceptionUtils.RpcRuntimeExceptionWrap(e.getCause());
        } catch (Exception e) {
            //一般为业务化异常 应不会出现中断异常
            throw ExceptionUtils.RpcRuntimeExceptionWrap(e);
        }
        return response;
    }

    @Override
    public boolean isAlive() {
        return channel.isOpen() && channel.isActive();
    }

    @Override
    public void close() throws RpcRuntimeException {
        try {
            channel.close();
        } catch (Exception e) {
            throw ExceptionUtils.RpcRuntimeExceptionWrap(e);
        }
    }

    @Override
    public void exchange(RpcExchange rpcExchange) {
        this.sync = rpcExchange;
    }


    @Override
    public RpcNode node() {
        return node;
    }

    public void setChannel(Channel channel) {
        this.channel = channel;
    }
}
